# The applied options are for CSV files. For other file types, these will be ignored.
col = ['participant_id' , 'activity_code' , 'timestamp', 'x', 'y', 'z']
raw_par_10_phone_accel = spark.read.format("csv") \
             .option("header", "false") \
             .option("inferSchema", "true") \
             .option("delimiter", ",") \
             .load("s3://humanactivity/wisdm-dataset/raw/phone/accel/data_1610_accel_phone.txt") \
             .toDF(*col)

display(raw_par_10_phone_accel)

raw_par_10_phone_accel.show(5)

raw_par_10_phone_accel.dtypes

from pyspark.sql.functions import regexp_replace
raw_par_10_phone_accel=raw_par_10_phone_accel.withColumn('z',regexp_replace('z',';',''))

from pyspark.sql.functions import col
from pyspark.sql.types import DoubleType

from pyspark.sql.types import DoubleType
raw_par_10_phone_accel = raw_par_10_phone_accel.withColumn("z", col("z").cast(DoubleType()))

raw_par_10_phone_accel.dtypes

from pyspark.sql.functions import unix_timestamp
raw_par_10_phone_accel = raw_par_10_phone_accel.withColumn('timestamp', col('timestamp').cast('timestamp'))

raw_par_10_phone_accel.dtypes

raw_par_10_phone_accel.show(5)

from pyspark.sql.functions import *
columns = ['participant_id','activity_code','timestamp', 'x','y','z']
for i in columns:
    print(i,raw_par_10_phone_accel.filter(raw_par_10_phone_accel[i].isNull()).count())

activity_codes_mapping = {'A': 'walking',
                          'B': 'jogging',
                          'C': 'stairs',
                          'D': 'sitting',
                          'E': 'standing',
                          'F': 'typing',
                          'G': 'brushing teeth',
                          'H': 'eating soup',
                          'I': 'eating chips',
                          'J': 'eating pasta',
                          'K': 'drinking from cup',
                          'L': 'eating sandwich',
                          'M': 'kicking soccer ball',
                          'O': 'playing catch tennis ball',
                          'P': 'dribbling basket ball',
                          'Q': 'writing',
                          'R': 'clapping',
                          'S': 'folding clothes'}

def activity_codes_mapping_udf(code):
    return activity_codes_mapping.get(code, 'unknown')

activity_color_map = {activity_codes_mapping['A']: 'lime',
                      activity_codes_mapping['B']: 'red',
                      activity_codes_mapping['C']: 'blue',
                      activity_codes_mapping['D']: 'orange',
                      activity_codes_mapping['E']: 'yellow',
                      activity_codes_mapping['F']: 'lightgreen',
                      activity_codes_mapping['G']: 'greenyellow',
                      activity_codes_mapping['H']: 'magenta',
                      activity_codes_mapping['I']: 'gold',
                      activity_codes_mapping['J']: 'cyan',
                      activity_codes_mapping['K']: 'purple',
                      activity_codes_mapping['L']: 'lightgreen',
                      activity_codes_mapping['M']: 'violet',
                      activity_codes_mapping['O']: 'limegreen',
                      activity_codes_mapping['P']: 'deepskyblue',   
                      activity_codes_mapping['Q']: 'mediumspringgreen',
                      activity_codes_mapping['R']: 'plum',
                      activity_codes_mapping['S']: 'olive'}

column_data=raw_par_10_phone_accel.select('x')
histogram = column_data.rdd.flatMap(lambda x: x).histogram(4)

# plot the histogram
import matplotlib.pyplot as plt
plt.hist(column_data.rdd.flatMap(lambda x: x).collect(), bins=50, color='green')
plt.show()

column_data=raw_par_10_phone_accel.select('y')
histogram = column_data.rdd.flatMap(lambda x: x).histogram(4)

# plot the histogram
import matplotlib.pyplot as plt
plt.hist(column_data.rdd.flatMap(lambda x: x).collect(), bins=50, color='green')
plt.show()

column_data=raw_par_10_phone_accel.select('z')
histogram = column_data.rdd.flatMap(lambda x: x).histogram(4)

# plot the histogram
import matplotlib.pyplot as plt
plt.hist(column_data.rdd.flatMap(lambda x: x).collect(), bins=50, color='green')
plt.show()

def show_accel_per_activity(device, df, act, interval_in_sec = None):
  ''' Plots acceleration time history per activity '''

  df1 = df.loc[df.activity_code == act].copy()
  df1.reset_index(drop = True, inplace = True)
  
  df1['duration'] = (df1['timestamp'] - df1['timestamp'].iloc[0])/1000000000 # nanoseconds --> seconds
  
  if interval_in_sec == None:
    ax = df1[:].plot(kind='line', x='duration', y=['x','y','z'], figsize=(25,7), grid = True) # ,title = act)
  else:
    ax = df1[:interval_in_sec*20].plot(kind='line', x='duration', y=['x','y','z'], figsize=(25,7), grid = True) # ,title = act)

  ax.set_xlabel('duration  (sec)', fontsize = 15)
  ax.set_ylabel('acceleration  (m/sec^2)',fontsize = 15)
  ax.set_title('Acceleration:   Device: ' + device + '      Activity:  ' +activity_codes_mapping[act], fontsize = 15)

raw_par_10_phone_accel_pan=raw_par_10_phone_accel.toPandas()

for key in activity_codes_mapping:
  show_accel_per_activity('Phone', raw_par_10_phone_accel_pan, key, 10)

col = ['participant_id' , 'activity_code' , 'timestamp', 'x', 'y', 'z']
raw_par_20_watch_accel = spark.read.format("csv") \
             .option("header", "false") \
             .option("inferSchema", "true") \
             .option("delimiter", ",") \
             .load("s3://humanactivity/wisdm-dataset/raw/watch/accel/data_1620_accel_watch.txt") \
             .toDF(*col)

display(raw_par_20_watch_accel)

from pyspark.sql.functions import regexp_replace
raw_par_20_watch_accel=raw_par_20_watch_accel.withColumn('z',regexp_replace('z',';',''))

from pyspark.sql.types import DoubleType
raw_par_20_watch_accel = raw_par_20_watch_accel.withColumn("z", col("z").cast(DoubleType()))

raw_par_20_watch_accel.dtypes

from pyspark.sql.functions import unix_timestamp
raw_par_20_watch_accel = raw_par_20_watch_accel.withColumn('timestamp', col('timestamp').cast('timestamp'))

raw_par_20_watch_accel.show(4)

raw_par_20_watch_accel.dtypes

from pyspark.sql.functions import *
columns = ['participant_id','activity_code','timestamp', 'x','y','z']
for i in columns:
    print(i,raw_par_20_watch_accel.filter(raw_par_20_watch_accel[i].isNull()).count())

column_data=raw_par_20_watch_accel.select('x')
histogram = column_data.rdd.flatMap(lambda x: x).histogram(4)

# plot the histogram
import matplotlib.pyplot as plt
plt.hist(column_data.rdd.flatMap(lambda x: x).collect(), bins=50, color='green')
plt.show()

column_data=raw_par_20_watch_accel.select('y')
histogram = column_data.rdd.flatMap(lambda x: x).histogram(4)

# plot the histogram
import matplotlib.pyplot as plt
plt.hist(column_data.rdd.flatMap(lambda x: x).collect(), bins=50, color='green')
plt.show()

column_data=raw_par_20_watch_accel.select('z')
histogram = column_data.rdd.flatMap(lambda x: x).histogram(4)

# plot the histogram
import matplotlib.pyplot as plt
plt.hist(column_data.rdd.flatMap(lambda x: x).collect(), bins=50, color='green')
plt.show()

raw_par_20_watch_accel_pan=raw_par_20_watch_accel.toPandas()

col = ['participant_id' , 'activity_code' , 'timestamp', 'x', 'y', 'z']
raw_par_35_phone_gyro = spark.read.format("csv") \
             .option("header", "false") \
             .option("inferSchema", "true") \
             .option("delimiter", ",") \
             .load("s3://humanactivity/wisdm-dataset/raw/phone/gyro/data_1635_gyro_phone.txt") \
             .toDF(*col)

display(raw_par_35_phone_gyro)

raw_par_35_phone_gyro=raw_par_35_phone_gyro.withColumn('z',regexp_replace('z',';',''))

from pyspark.sql.functions import col
raw_par_35_phone_gyro = raw_par_35_phone_gyro.withColumn("z", col("z").cast(DoubleType()))
raw_par_35_phone_gyro.dtypes

1
column_data=raw_par_35_phone_gyro.select('x')

2
histogram = column_data.rdd.flatMap(lambda x: x).histogram(4)

3
 

4
# plot the histogram

5
import matplotlib.pyplot as plt

6
plt.hist(column_data.rdd.flatMap(lambda x: x).collect(), bins=50, color='green')

7
plt.show()

8
 

9
 

10
column_data=raw_par_35_phone_gyro.select('y')

11
histogram = column_data.rdd.flatMap(lambda x: x).histogram(4)

12
 

13
# plot the histogram

14
import matplotlib.pyplot as plt

15
plt.hist(column_data.rdd.flatMap(lambda x: x).collect(), bins=50, color='green')

16
plt.show()

17
 

18
 

19
 

20
column_data=raw_par_35_phone_gyro.select('z')

21
histogram = column_data.rdd.flatMap(lambda x: x).histogram(4)

22
 

23
# plot the histogram

24
import matplotlib.pyplot as plt

25
plt.hist(column_data.rdd.flatMap(lambda x: x).collect(), bins=50, color='green')

26
plt.show()

raw_par_35_phone_gyro_pan=raw_par_35_phone_gyro.toPandas()

def show_ang_velocity_per_activity(device, df, act, interval_in_sec = None):
  ''' Plots angular volocity time history per activity '''

  df1 = df.loc[df.activity_code == act].copy()
  df1.reset_index(drop = True, inplace = True)

  df1['duration'] = (df1['timestamp'] - df1['timestamp'].iloc[0])/1000000000 # nanoseconds --> seconds

  if interval_in_sec == None:
    ax = df1[:].plot(kind='line', x='duration', y=['x','y','z'], figsize=(25,7), grid = True) # ,title = act)
  else:
    ax = df1[:interval_in_sec*20].plot(kind='line', x='duration', y=['x','y','z'], figsize=(25,7), grid = True) # ,title = act)

  ax.set_xlabel('duration  (sec)', fontsize = 15)
  ax.set_ylabel('angular velocity  (rad/sec)',fontsize = 15)
  ax.set_title('Angular velocity:  Device: ' + device + '      Activity:  ' +activity_codes_mapping[act] , fontsize = 15)

for key in activity_codes_mapping:
  show_ang_velocity_per_activity('Phone', raw_par_35_phone_gyro_pan, key)

col = ['participant_id' , 'activity_code' , 'timestamp', 'x', 'y', 'z']
raw_par_35_watch_gyro = spark.read.format("csv") \
             .option("header", "false") \
             .option("inferSchema", "true") \
             .option("delimiter", ",") \
             .load("s3://humanactivity/wisdm-dataset/raw/watch/gyro/data_1635_gyro_watch.txt") \
             .toDF(*col)

display(raw_par_35_watch_gyro)

1
raw_par_35_watch_gyro=raw_par_35_watch_gyro.withColumn('z',regexp_replace('z',';',''))

raw_par_35_watch_gyro_pan=raw_par_35_watch_gyro.toPandas()

for key in activity_codes_mapping:
  show_ang_velocity_per_activity('Watch', raw_par_35_watch_gyro_pan, key)

from pyspark.sql.functions import col
from pyspark.sql.types import DoubleType
for coli in all_phone_accel.columns:
    if coli == 'ACTIVITY':
        continue
    else:
        all_phone_accel = all_phone_accel.withColumn(coli, col(coli).cast(DoubleType()))
        

import matplotlib.pyplot as plt
import pandas as pd

# Group by activity and count the rows
activity_counts = all_phone_accel.groupBy('ACTIVITY') \
    .count() \
    .orderBy('count', ascending=False)

# Convert to Pandas DataFrame
activity_counts_pd = activity_counts.toPandas()

# Plot the results
_ = activity_counts_pd.plot(kind='bar', x='ACTIVITY', y='count',
                            figsize=(15, 5), color='purple',
                            title='row count per activity',
                            legend=True, fontsize=15)
plt.show()

# Group by activity and count the rows
activity_counts = all_phone_accel.groupBy('PARTICIPANT') \
    .count() \
    .orderBy('count', ascending=False)

# Convert to Pandas DataFrame
activity_counts_pd = activity_counts.toPandas()

# Plot the results
_ = activity_counts_pd.plot(kind='bar', x='PARTICIPANT', y='count',
                            figsize=(15, 5), color='purple',
                            title='row count per PARTICIPANT',
                            legend=True, fontsize=15)
plt.show()

all_phone_accel[['XABSOLDEV', 'YABSOLDEV','ZABSOLDEV','XSTANDDEV', 'YSTANDDEV', 'ZSTANDDEV', 'XVAR', 'YVAR', 'ZVAR']].show(4)

all_phone_accel = all_phone_accel.drop('XSTANDDEV', 'YSTANDDEV', 'ZSTANDDEV', 'XVAR', 'YVAR', 'ZVAR')

all_phone_accel = all_phone_accel.drop('PARTICIPANT')

X_train = spark.read.csv("s3://humanactivity/Train_test_spllit/X_train",header=True)
X_test = spark.read.csv("s3://humanactivity/Train_test_spllit/X_test",header=True)
y_train=spark.read.csv("s3://humanactivity/Train_test_spllit/y_train",header=True)
y_test= spark.read.csv("s3://humanactivity/Train_test_spllit/y_test",header=True)

X_test.count(),X_train.count()

y_test.count(),y_train.count()

par_23_df = spark.read.csv("s3://humanactivity/Cluster_diagram/par_23_df",header=True)

import matplotlib.pyplot as plt
from sklearn.manifold import TSNE
import pandas as pd

# Convert the PySpark DataFrame to a Pandas DataFrame
par_23_pd_df = par_23_df.toPandas()

# Extract the features and the target variable from the Pandas DataFrame
yy = par_23_pd_df['ACTIVITY']
XX = par_23_pd_df.drop(['ACTIVITY','PARTICIPANT','ACT','XSTANDDEV','YSTANDDEV','ZSTANDDEV','XVAR','YVAR','ZVAR'], axis = 1)

# Apply t-SNE to reduce the dimensionality of the feature space to 2 dimensions
tsne = TSNE(n_components=2, random_state=300)
X_2d = tsne.fit_transform(XX)

# Plot the t-SNE visualization using Matplotlib
target_ids = tuple(activity_codes_mapping.keys())

plt.figure(figsize=(10, 10))
colors = 'lime', 'red', 'blue', 'orange', 'yellow', 'lightgreen', 'greenyellow', 'magenta', 'gold', 'cyan', 'purple', 'lightgreen', 'violet', 'limegreen', 'deepskyblue', 'mediumspringgreen', 'plum', 'olive'

for i, c, label in zip(target_ids, colors, tuple(activity_codes_mapping.values())):
    plt.scatter(X_2d[yy == i, 0], X_2d[yy == i, 1], c=c, label=label)

plt.legend()
plt.show()

par_35_df = spark.read.csv("s3://humanactivity/Cluster_diagram/par_35_df",header=True)

# Convert the PySpark DataFrame to a Pandas DataFrame
par_35_pd_df = par_35_df.toPandas()

# Extract the features and the target variable from the Pandas DataFrame
yy = par_23_pd_df['ACTIVITY']
XX = par_23_pd_df.drop(['ACTIVITY','PARTICIPANT','ACT','XSTANDDEV','YSTANDDEV','ZSTANDDEV','XVAR','YVAR','ZVAR'], axis = 1)

# Apply t-SNE to reduce the dimensionality of the feature space to 2 dimensions
tsne = TSNE(n_components=2, random_state=300)
X_2d = tsne.fit_transform(XX)

# Plot the t-SNE visualization using Matplotlib
target_ids = tuple(activity_codes_mapping.keys())

plt.figure(figsize=(10, 10))
colors = 'lime', 'red', 'blue', 'orange', 'yellow', 'lightgreen', 'greenyellow', 'magenta', 'gold', 'cyan', 'purple', 'lightgreen', 'violet', 'limegreen', 'deepskyblue', 'mediumspringgreen', 'plum', 'olive'

for i, c, label in zip(target_ids, colors, tuple(activity_codes_mapping.values())):
    plt.scatter(X_2d[yy == i, 0], X_2d[yy == i, 1], c=c, label=label)

plt.legend()
plt.show()

par_40_df = spark.read.csv("s3://humanactivity/Cluster_diagram/par_40_df",header=True)

# Convert the PySpark DataFrame to a Pandas DataFrame
par_40_pd_df = par_40_df.toPandas()

# Extract the features and the target variable from the Pandas DataFrame
yy = par_23_pd_df['ACTIVITY']
XX = par_23_pd_df.drop(['ACTIVITY','PARTICIPANT','ACT','XSTANDDEV','YSTANDDEV','ZSTANDDEV','XVAR','YVAR','ZVAR'], axis = 1)

# Apply t-SNE to reduce the dimensionality of the feature space to 2 dimensions
tsne = TSNE(n_components=2, random_state=300)
X_2d = tsne.fit_transform(XX)

# Plot the t-SNE visualization using Matplotlib
target_ids = tuple(activity_codes_mapping.keys())

plt.figure(figsize=(10, 10))
colors = 'lime', 'red', 'blue', 'orange', 'yellow', 'lightgreen', 'greenyellow', 'magenta', 'gold', 'cyan', 'purple', 'lightgreen', 'violet', 'limegreen', 'deepskyblue', 'mediumspringgreen', 'plum', 'olive'

for i, c, label in zip(target_ids, colors, tuple(activity_codes_mapping.values())):
    plt.scatter(X_2d[yy == i, 0], X_2d[yy == i, 1], c=c, label=label)

plt.legend()
plt.show()

1
import pandas as pd

2
import matplotlib.pyplot as plt

3
from pyspark.sql.functions import count

4
 

5
counts = y_train.groupBy('Y').agg(count('*').alias('count'))

6
activity_counts_pd = counts.toPandas()

7
 

8
activity_counts_pd.plot(kind='bar', x='Y', y='count', color='red', figsize=(15,5), legend=False, fontsize=15)

9
plt.title('Row count per activity', fontsize=15)

10
plt.show()

11
 

counts = y_test.groupBy('ACTIVITY').agg(count('*').alias('count'))
activity_counts_pd = counts.toPandas()

activity_counts_pd.plot(kind='bar', x='ACTIVITY', y='count', color='red', figsize=(15,5), legend=False, fontsize=15)
plt.title('Row count per activity', fontsize=15)
plt.show()

from sklearn.neighbors import KNeighborsClassifier

knn_classifier = KNeighborsClassifier()

my_param_grid = {'n_neighbors': [5, 10, 20], 'leaf_size': [20, 30, 40]}

from sklearn.model_selection import cross_val_score
from sklearn.model_selection import StratifiedShuffleSplit
from sklearn.model_selection import GroupKFold

my_cv = StratifiedShuffleSplit(n_splits=5, train_size=0.7, test_size=0.3)

from sklearn.model_selection import GridSearchCV
knn_model_gs = GridSearchCV(estimator = knn_classifier, 
                            param_grid = my_param_grid,
                            cv = my_cv, 
                            scoring ='accuracy')